package com.bst.datax.executor.service.jobhandler;

import com.bst.common.utils.HdpDateUtil;
import com.bst.common.utils.sql.SqlUtil;
import com.bst.datatx.core.biz.model.JobParam;
import com.bst.datatx.core.biz.model.ReturnT;
import com.bst.datatx.core.biz.model.TriggerParam;
import com.bst.datatx.core.handler.IJobHandler;
import com.bst.datatx.core.handler.annotation.JobHandler;
import com.bst.datatx.core.log.JobLogger;
import com.bst.datax.executor.service.logparse.LogStatistics;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.stereotype.Component;

import java.sql.Connection;
import java.sql.DriverManager;
import java.sql.SQLException;
import java.sql.Statement;
import java.text.ParseException;
import java.text.SimpleDateFormat;
import java.util.Calendar;
import java.util.Date;

/**
 * DataX任务运行
 *
 * @author jingwk 2019-11-16
 */

@JobHandler(value = "executorDbHandler")
@Component
public class ExecutorDbHandler extends IJobHandler {

    @Value("${datax.executor.jsonpath}")
    private String jsonPath;

    @Value("${datax.pypath}")
    private String dataXPyPath;


    /**
     * 防止一次性插入数据量过大，加入按月执行操作
     * @param trigger
     * @return
     */
    @Override
    public ReturnT<String> execute(TriggerParam trigger) {
        Connection conn = null;
        Statement stmt = null;
        Date beginDate = new Date();
        Integer originalLevel = null;
        JobParam jobParam = trigger.getJobParam();
        String tableName = jobParam.getTableName();
        String dbType="";
        try {
            String sql = jobParam.getInsertSql();
            conn = DriverManager.getConnection(jobParam.getUrl(),jobParam.getUsername(),jobParam.getPassword());
            originalLevel = conn.getTransactionIsolation();
            conn.setTransactionIsolation(Connection.TRANSACTION_READ_UNCOMMITTED);
            String dayBegin = jobParam.getDayBegin(),dayEnd=jobParam.getDayEnd();
            dbType = conn.getMetaData().getDatabaseProductName().replace(" ","").toUpperCase();
            JobLogger.log("<br>"+dbType+",数据提取开始执行时间：" +HdpDateUtil.FORMAT_YYYY_MM_DD_HHMMSS.format(beginDate));
            LogStatistics statistics = new LogStatistics();
            DataBatchInsertHandler insertHandler = new DataBatchInsertHandler(jobParam,statistics);
            insertHandler.batchDelete(conn,tableName,dbType,dayBegin,dayEnd);
            Date endDate = new Date();
            long seconds = (endDate.getTime()-beginDate.getTime())/1000;
            JobLogger.log("<br>执行删除耗时：" + seconds+"s,开始执行插入");
            if(dbType.contains("MYSQL")) {
                insertHandler.executeAll(conn);
            } else {
                insertHandler.execute(conn);
            }
            endDate = new Date();
            seconds = (endDate.getTime()-beginDate.getTime())/1000;
            statistics.setTaskStartTime(HdpDateUtil.FORMAT_YYYY_MM_DD_HHMMSS.format(beginDate));
            statistics.setTaskEndTime(HdpDateUtil.FORMAT_YYYY_MM_DD_HHMMSS.format(endDate));
            statistics.setTaskTotalTime(seconds+"s");
            statistics.setTaskRecordWritingSpeed(seconds>0?(statistics.getTaskRecordReaderNum()/seconds)+"":"0");
            if(statistics.getTaskRecordWriteFailNum()!=0) {
                return new ReturnT<>(IJobHandler.FAIL.getCode(), statistics.toString());
            } else {
                return new ReturnT<>(200, statistics.toString());
            }
        } catch (Exception e) {
            e.printStackTrace();
            return new ReturnT<>(IJobHandler.FAIL.getCode(), "error{msg="+e.getMessage()+"}");
        } finally {
            try {
                // DataBatchInsertHandler.enableIndexes(conn,tableName,dbType);
                if(originalLevel!=null) {
                    conn.setTransactionIsolation(originalLevel);
                }
                if(stmt!=null) {
                    stmt.close();
                }
                if(conn!=null) {
                    conn.close();
                }
            } catch (SQLException throwables) {
                // throwables.printStackTrace();
            }
        }
    }

    public static void main(String[] args) {
        String sql = "select * from table where date>'${DT_BEGIN}' and date<'${DT_END}' ";
        String dayBegin="2023-01-01",dayEnd="2024-01-01";
        try {
            SimpleDateFormat format = HdpDateUtil.FORMAT_YYYY_MM_DD;
            Date dateBegin = format.parse(dayBegin);
            Date dateEnd = format.parse(dayEnd);
            Long days =HdpDateUtil.diffDays(dateEnd,dateBegin);
            if(days<35) {
                System.out.println(SqlUtil.buildSql(sql,dayBegin,dayEnd,"sqlserver"));
            } else {
                Date nextMonth = dateBegin;
                while (nextMonth.getTime()<dateEnd.getTime()) {
                    nextMonth =HdpDateUtil.calculate(dateBegin, Calendar.MONTH,1);
                    dayEnd = format.format(nextMonth);
                    System.out.println(SqlUtil.buildSql(sql,dayBegin,dayEnd,"sqlserver"));
                    dateBegin = nextMonth;
                    dayBegin = format.format(dateBegin);
                }
            }
        } catch (ParseException e) {
            e.printStackTrace();
        }
    }
}
